package wc;

import java.io.IOException;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class mba{
    public static class MBAMapper extends Mapper<Object,Text,Text,IntWritable>{
        final IntWritable one = new IntWritable(1);
        private int k;
        public void setup(Context context){
            k = Integer.parseInt(context.getConfiguration().get("k"));
        }
     
        public static List<List<String>> findSortedCombanations( List<String> sortedElements, int n){
            List<List<String>> fResult = new ArrayList<List<String>>();
            findSortedCombanations(sortedElements, n, fResult);
            return fResult;
        }

        public static List<List<String>> findSortedCombanations(List<String> sortedElements, int n, List<List<String>> fResult){
            List<List<String>> result = new ArrayList<List<String>>();
            if(n==0){
                result.add(new ArrayList<String>());
                fResult.add(new ArrayList<String>());
                return result;
            }
            List<List<String>> combinations = findSortedCombanations(sortedElements, n - 1, fResult);

            for(List<String> combination : combinations){
                int Index = -1;
                if(combination.size() > 0) {
                    String lastElement = combination.get(combination.size() - 1);
                    Index = sortedElements.indexOf(lastElement);
                }
                for(int i = Index + 1; i < sortedElements.size(); i++){
                    String element = sortedElements.get(i);
                    List<String> cloneList = new ArrayList<String>(combination);
                    cloneList.add(element);
                    result.add(cloneList);
                }
            }
            fResult.addAll(result);
            return result;
        }

        @Override
        public void map(Object key, Text values, Context context) throws IOException, InterruptedException {
            String[] goods = values.toString().split(" ");
            List<List<String>> results = findSortedCombanations(Arrays.asList(goods), k);
            for (List<String> result : results)
                if (result.size() == k){
                    String str = "(";
                    for (int i = 0; i < k; i++)
                        str = str + result.get(i) + ",";
                    context.write(new Text(str.substring(0,str.length() - 1) + ")"),one);
                }
        }
    }
    public static class MBAReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
        private int min;
        List<List<String>> list = new ArrayList<List<String>>();
        public void setup(Context context){
            min = Integer.parseInt(context.getConfiguration().get("min"));
        }
        public void reduce(Text key, Iterable<IntWritable> values, Context context)throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            if(result.get() >= min){
                List<String> sublist = new ArrayList<String>();
                sublist.add(key.toString());
                sublist.add(result.toString());
                list.add(sublist);
            }
        }
        public void cleanup(Context context) throws IOException, InterruptedException{
            for (int i = 0; i < list.size() - 1; i++)
                for(int j = i + 1; j < list.size(); j++)
                    if (Integer.parseInt(list.get(i).get(1)) < Integer.parseInt(list.get(j).get(1)))
                    {
                        List<String> tmp =list.get(i);
                        list.set(i, list.get(j));
                        list.set(j, tmp);
                    }
            for(List<String> sublist : list){
                    IntWritable new_value = new IntWritable();
                    new_value.set(Integer.parseInt(sublist.get(1)));
                    context.write(new Text(sublist.get(0)) , new_value);
            }
        }
    }
    
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        conf.set("mapreduce.output.textoutputformat.separator", ",");
        conf.set("k", args[2]);
        conf.set("min", args[3]);
        Job job = Job.getInstance(conf, "mba 1.0");
        job.setJarByClass(mba.class);
        job.setMapperClass(MBAMapper.class);
        job.setCombinerClass(MBAReducer.class);
        job.setReducerClass(MBAReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

